package club.hicode.daydayup.zookeeper.curator;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Consumer;

/**
 * 分布式锁技术预研
 *
 * <li>﻿InterProcessMutex</li>
 * <li>﻿InterProcessReadWriteLock</li>
 * <li>﻿InterProcessSemaphoreV2</li>
 *
 * @author Liu Chunfu
 * @date 2018-06-07 下午5:07
 **/
public class DistributeLock {

    /**
     * 链接字符串
     */
    private static final String CONNECT_STR = "127.0.0.1:2181";

    /**
     * 锁定的根路径
     */
    private static final String LOCK = "/lock";

    @Test
    public void testMutex() {
        CuratorFramework client = getClient();
        InterProcessMutex mutex = new InterProcessMutex(client, LOCK);
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 5; i++) {
            executorService.execute(() -> {
                try {
                    mutex.acquire();
                    //Participant 参与者
                    System.out.println("开始做....");
                    ThreadUtil.sleep(RandomUtil.randomInt(5), TimeUnit.SECONDS);
                    System.out.println("事情结束...");
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    try {
                        mutex.release();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        ThreadUtil.sleep(5, TimeUnit.MINUTES);
    }


    @Test
    public void test() {
        //创建zookeeper客户端
        CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", new ExponentialBackoffRetry(1000, 3));
        client.start();

        //指定锁路径
        String lockPath = "/zkLockRoot/lock_1";
        //创建锁，为可重入锁，即是获锁后，还可以再次获取，本例以此为例
        InterProcessMutex lock = new InterProcessMutex(client, lockPath);
//        //创建锁，为不可重入锁，即是获锁后，不可以再次获取，这里不作例子，使用和重入锁类似
//        InterProcessSemaphoreMutex lock = new InterProcessSemaphoreMutex(client, lockPath);

        ExecutorService executor = Executors.newCachedThreadPool();
        Consumer<InterProcessMutex> consumer = (InterProcessMutex typeLock) -> {
            try {
                List<Callable<String>> callList = new ArrayList<>();
                Callable<String> call = () -> {
                    try {
                        //获取锁
                        typeLock.acquire();
                        System.out.println(Thread.currentThread() + "  acquire read lock");
                    } catch (Exception e) {
                    } finally {
                        //释放锁
                        typeLock.release();
                        System.out.println(Thread.currentThread() + "  release read lock");
                    }
                    return "true";
                };
                //5个并发线程
                for (int i = 0; i < 5; i++) {
                    callList.add(call);
                }
                List<Future<String>> futures = executor.invokeAll(callList);
            } catch (Exception e) {

            }
        };

        //分布式锁测试
        consumer.accept(lock);

        executor.shutdown();
    }


    /**
     * 获取操作基础框架
     *
     * @return
     */
    private CuratorFramework getClient() {
        CuratorFramework framework = CuratorFrameworkFactory.newClient(CONNECT_STR, new RetryNTimes(3, 3000));
        framework.start();
        return framework;
    }


}
